package com.zhao.demo.unbound.demo04_transformation.sample02_reduce

import com.zhao.demo.Raytek
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * Description: 求出安放在北京西站的红外测温仪迄今为止所测得的最早的时间,以及测得的最高的温度<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/8 15:58
 *
 * @author 柒柒
 * @version : 1.0
 */

object ReduceDemo {
  def main(args: Array[String]): Unit = {
    //1.环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //2.实时读取流数据,计算,并显示结果
    import org.apache.flink.api.scala._

    env.socketTextStream("node01",7776)
      .map(perInfo =>{
        val arr = perInfo.split(",")
        val id = arr(0).trim
        val temperature = arr(1).trim.toDouble
        val name = arr(2).trim
        val timestamp = arr(3).trim.toLong
        val location = arr(4).trim
        Raytek(id,temperature,name,timestamp,location)
      }).keyBy("id") //根据红外测温仪的id进行分组
    //求各个分组中最早的时间,以及最高的温度
      .reduce((before: Raytek, after: Raytek) =>{
        val maxTemperture: Double = Math.max(before.temperature, after.temperature)
        val minTimestamp: Long = Math.min(before.timestamp, after.timestamp)
        Raytek(before.id,maxTemperture,"",minTimestamp,"")
      }).print("当前红外测温仪迄今为止测得的最高的体温,最早的时间->")

    //3.启动
    env.execute()
  }
}










